package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.commont.Constant;
import com.atguigu.gmall.realtime.function.KwSplit;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * @Author lzc
 * @Date 2023/4/27 09:00
 */
public class Dws_01_DwsTrafficSourceKeywordPageViewWindow extends BaseSQLApp {
    public static void main(String[] args) {
        new Dws_01_DwsTrafficSourceKeywordPageViewWindow().init(
            40001,
            2,
            "Dws_01_DwsTrafficSourceKeywordPageViewWindow"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 建立动态表与页面日志关联
        tEnv.executeSql("create table page(" +
                            " page map<string, string>, " +
                            " ts bigint, " +
                            " et as to_timestamp_ltz(ts, 3)," +
                            " watermark for et as et - interval '3' second " +
                            ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE, "Dws_01_DwsTrafficSourceKeywordPageViewWindow"));
        
        // 2. 过滤出搜索记录, 取出关键词
        // 搜索记录: 1. last_page_id = home && item_type=keyword && item != null
        // 搜索记录: 2. last_page_id = search && item_type=keyword && item != null
        // 综合: (last_page_id=home or last_page_id = search) && item_type=keyword && item != null
        // 综合: (last_page_id in (home, search)) && item_type=keyword && item != null
        Table kwTable = tEnv.sqlQuery("select " +
                                          " page['item'] kw, " +
                                          " et " +
                                          "from page " +
                                          "where page['last_page_id'] in ('home', 'search') " +
                                          "and page['item_type'] ='keyword' " +
                                          "and page['item'] is not null ");
        tEnv.createTemporaryView("kw_table", kwTable);
        
        // 3. 对关键词进行分词: 需要自定义制表函数
        // 3.1 自定义制表函数
        tEnv.createTemporaryFunction("kw_split", KwSplit.class);
        // 3.2 使用自定义函数,把关键词进行分词
        Table keywordTable = tEnv.sqlQuery("select " +
                                               " keyword, " +
                                               " et " +
                                               "from kw_table " +
                                               "join lateral table(kw_split(kw)) on true");
        tEnv.createTemporaryView("keyword_table", keywordTable);
        // 3. 开窗聚合
        Table result = tEnv.sqlQuery(
            "select " +
                " date_format(window_start, 'yyyy-MM-dd HH:mm:ss') stt, " +
                " date_format(window_end, 'yyyy-MM-dd HH:mm:ss') edt, " +
                " keyword , " +
                " count(*) keyword_count, " +
                " UNIX_TIMESTAMP() * 1000 as ts  " + // 计数: count(*) count(id) sum(1)
                "from table(tumble(table keyword_table,  descriptor(et), interval '5' second )) " +
                "group by window_start, window_end, keyword");
        
        
        // 4.聚合后的结果写出到 doris 中
        
        
        
//        // 4. 写出
//        // 专门的 clickhouse sink 就好了
//        // 自定义 sink: 用流做 sink
//        SingleOutputStreamOperator<KeywordBean> stream = tEnv
//            .toRetractStream(result, KeywordBean.class)
//            .filter(t -> t.f0)
//            .map(t -> t.f1);
//
//        stream.addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_keyword_page_view_window", KeywordBean.class));
//
    
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    
    
    }
}
/*
0-5  华为 手机  1
0-5  华为手机  1
0-5  苹果手机  1
0-5  小米手机  1
---
分词:
    华为 手机
    华为手机
    苹果手机
    小米手机
  =>
    华为
    手机
    华为
    手机
    苹果
    手机
    小米
    手机
    
    列转行: 制表函数: TableFunction

------
 dws层设计思路:
    小窗口汇总
 1. 读取数据:
        页面日志, 找到搜索记录
        
 2. 取出搜索关键词
 
 5. 开窗聚合
        三种:
            grouped window
                tvf: 是用来替换 grouped window
                    tumble 滚动
                    hop    滑动
                    cumulate 累积
            
            over: topN
            
 5. 数据写出到数据库中:
     clickhouse


最终结果:
0-5  手机  10
0-5  电脑  20
5-10 ...  ...

...
 */